package broker

import (
	"context"
	"strings"

	"gitee.com/go-course/go9/projects/devcloud/maudit/apps/audit"
	"github.com/segmentio/kafka-go"
)

const (
	DEFAULT_TOPIC = "audit_log"
)

func NewDefaultClient(brokers string) *Client {
	return NewClient(brokers, DEFAULT_TOPIC)
}

func NewClient(brokers, topic string) *Client {
	adress := strings.Split(brokers, ",")
	w := &kafka.Writer{
		Addr:                   kafka.TCP(adress...),
		Topic:                  topic,
		Balancer:               &kafka.LeastBytes{},
		AllowAutoTopicCreation: false,
	}
	return &Client{
		brokerAddress: adress,
		writer:        w,
	}
}

type Client struct {
	brokerAddress []string
	writer        *kafka.Writer
}

func (c *Client) Close() error {
	return c.writer.Close()
}

// 发送审计日志
func (c *Client) SendAuditLog(ctx context.Context, al audit.AuditLog) error {
	// 构建一个kafka message
	m := kafka.Message{
		Key:   []byte(al.ServiceId),
		Value: al.ToJsonByte(),
	}

	return c.writer.WriteMessages(ctx, m)
}
